
package com.lee.source; 
 
import com.lee.database.HBaseConnPool; 
import com.lee.logger.Logger; 
import com.lee.utils.ConstantsUtils; 
import com.lee.utils.CountsUtils; 
import org.apache.commons.collections.CollectionUtils; 
import org.apache.commons.lang.StringUtils; 
import org.apache.hadoop.hbase.*; 
import org.apache.hadoop.hbase.client.*; 
import org.apache.hadoop.hbase.util.Bytes; 
 
import java.io.IOException; 
import java.util.ArrayList; 
import java.util.List; 
import java.util.Map; 
 
public class HbaseSource { 
    public static String getCrowdInfo(String key) throws IOException {
 
//        Table table = HBaseConnPool.getInstance().getConnection().getTable(TableName.valueOf("marketing:crowd_portrait"));
        Table table = HBaseConnPool.getInstance().getConnection().getTable(TableName.valueOf(ConstantsUtils.PORTRAIT_TABLE));
        Get get  = new Get(Bytes.toBytes(key));
        get.addFamily(Bytes.toBytes("tag"));
        Result result = table.get(get);
 
        for(Cell kv:result.rawCells()){
            String family =Bytes.toString(CellUtil.cloneFamily(kv));
            String qualifier =Bytes.toString(CellUtil.cloneQualifier(kv));
            String stringValue=Bytes.toString(CellUtil.cloneValue(kv));
            System.out.println(family +":"+ qualifier +"	"+stringValue);
        }
 
        return null;
 
 
    }
 
 
    public static ArrayList<String> getCrowdInfoMap(String key,String activityCode) throws IOException {
//        Table table = HBaseConnPool.getInstance().getConnection().getTable(TableName.valueOf("marketing:crowd_portrait"));
        Table table = HBaseConnPool.getInstance().getConnection().getTable(TableName.valueOf(ConstantsUtils.PORTRAIT_TABLE));
        if(StringUtils.isEmpty(key)){
            return new ArrayList<>();
        }
        Get get  = new Get(Bytes.toBytes(key));
        get.addFamily(Bytes.toBytes("tag"));
        Result result = table.get(get);
        ArrayList<String> values = new ArrayList<>();
        long activityNum = CountsUtils.getActivityValueByKey(activityCode) + 1;
        CountsUtils.setActivityValueByKey(activityCode,String.valueOf(activityNum));
        CountsUtils.setLaplaceActivityValueByKey(activityCode,activityNum);
        for(Cell kv:result.rawCells()){
            String qualifier =Bytes.toString(CellUtil.cloneQualifier(kv));
            String stringValue=Bytes.toString(CellUtil.cloneValue(kv));
            if("1".equals(stringValue)){
                values.add(qualifier);
                long tagNum = CountsUtils.getTagValueByKey(qualifier) + 1;
                CountsUtils.setTagValueByKey(qualifier,String.valueOf(tagNum));
                CountsUtils.setLaplaceTagValueByKey(qualifier,tagNum);
            }
        }
 
        return values;
 
 
    }
 
 
    public static String get(String key,String tableName,String family,String qualifier) throws IOException{
        Table table = HBaseConnPool.getInstance().getConnection().getTable(TableName.valueOf(tableName));
        Get get  = new Get(Bytes.toBytes(key));
        get.addColumn(Bytes.toBytes(family),Bytes.toBytes(qualifier));
        Result result = table.get(get);
        List<Cell> cells = result.listCells();
        if(CollectionUtils.isEmpty(cells)){
            return "";
        }
        String value = new String(CellUtil.cloneValue(cells.get(0)));
        return value;
    }
    public static void put(String key, String tableName, String family, Map<String,String> values) throws IOException {
        Table table = HBaseConnPool.getInstance().getConnection().getTable(TableName.valueOf(tableName));
        Put put = new Put(Bytes.toBytes(key));
        for(String qualifier:values.keySet()){
            put.addColumn(family.getBytes(),qualifier.getBytes(),values.get(qualifier).getBytes());
        }
        table.put(put);
 
    }
    public static void put(String key, String tableName, String family,String qualifier, String value) throws IOException {
        Table table = HBaseConnPool.getInstance().getConnection().getTable(TableName.valueOf(tableName));
        Put put = new Put(Bytes.toBytes(key));
        put.addColumn(family.getBytes(),qualifier.getBytes(),value.getBytes());
        table.put(put);
 
    }
 
    public static void delete(String key, String tableName) throws IOException {
        Table table = HBaseConnPool.getInstance().getConnection().getTable(TableName.valueOf(tableName));
        Delete delete = new Delete(key.getBytes());
        table.delete(delete);
    }
    public static void truncate(String tableName) throws IOException {
        try{
            Admin admin = HBaseConnPool.getInstance().getConnection().getAdmin();
            admin.disableTable(TableName.valueOf(tableName));
            admin.truncateTable(TableName.valueOf(tableName),true);
            admin.enableTable(TableName.valueOf(tableName));
 
        } catch (TableNotDisabledException e){
            Logger.error("HbaseSource truncate -> table ",tableName);
        }
    }
 
 
    public static void scan(String tableName) throws IOException {
        Table table = HBaseConnPool.getInstance().getConnection().getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        ResultScanner  scanner = table.getScanner(scan);
        for (Result res : scanner) {
            Logger.debug(res);
 
        }
 
        scanner.close();
 
 
 
    }
 
    public static List<String> getAllKey(String tableName) throws IOException {
 
        Table table = HBaseConnPool.getInstance().getConnection().getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        ResultScanner  scanner = table.getScanner(scan);
        List<String> rows = new ArrayList<>();
        for (Result res : scanner) {
            String row =  Bytes.toString(res.getRow());
            rows.add(row);
        }
 
        return rows;
 
 
    }
 
} 
